package com.xzx.consume;

import com.xzx.common.bean.CallLog;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * 从kafka消费数据写到hbase中
 *
 * @author xinzhixuan
 * @version 1.0
 * @date 2024/8/10 13:41
 */
public class AppConsumer {
    public static void main(String[] args) throws IOException {

        //1、先创建hbase表
        HbaseDao hbaseDao = new HbaseDao();
//        hbaseDao.createTable();
        //2、开始写入数据

        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.25.129.142:9092,172.25.129.133:9092,172.25.129.114:9092");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "dianxin-consumer");

        try (KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties)) {
            kafkaConsumer.subscribe(Collections.singletonList("xzx_calllog"));
            while (true) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(record.key() + "==>" + record.value());
                    String[] split = record.value().split("\t");
                    CallLog callLog = new CallLog();
                    callLog.setCaller(split[0]);
                    callLog.setCallee(split[1]);
                    callLog.setCallStartTime(split[2]);
                    callLog.setDuration(Integer.parseInt(split[3]));
                    hbaseDao.put(callLog);
                }
            }
        }

    }
}